package com.zw.kafka;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/**
 * @author morningstar
 * @date 2021/11/30
 * @desc
 */
public class MySimpleProducer1 {

    public final static String TOPIC_NAME = "my-replicated-topic";

    public static void main(String[] args) throws Exception {

        // 1.设置参数
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "tajlmt0law.cogiot.net:39092,tajlmt0law.cogiot.net:39093,tajlmt0law.cogiot.net:39094");

        // 把发送的key从字符串序列化为字节数组
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // 把发送消息value从字符串序列化为字节数组
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // 2.创建⽣产消息的客户端，传⼊参数
        Producer<String, String> producer = new KafkaProducer<String, String>(props);

        // 3.创建消息
        /**
         * 1.String topic: 主题
         * 2.Integer partition: 指定分区发送
         * 3.K key: 决定往哪个分区发送消息 计算公式： hash(key)%partitionNum
         * 4.V value: 具体要发送的消息内容
         */
        ProducerRecord<String, String> producerRecord = new ProducerRecord<>(TOPIC_NAME, 0,"hello", "Hello Kafka!");

        // 4.发送消息,得到消息发送的元数据并输出
        RecordMetadata metadata = producer.send(producerRecord).get();
        System.out.println("同步方式发送消息结果：topic-" + metadata.topic() + "|partition-" + metadata.partition() + "|offset-" + metadata.offset());


    }
}
